package com.yinhai.redis.cluster.event;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import com.yinhai.redis.cluster.event.filter.EventFilter;

/**
 * 事件总线
 * @author dreamlee
 */
public class EventBus {
	
	private static EventBus instance;
	
	private Map<EventFilter, List<EventListener>> listeners = Collections
			.synchronizedMap(new HashMap<EventFilter, List<EventListener>>());
	
	private final AllEventFilter allFilter = new AllEventFilter();
	
	private long timeout=5000;
	
	private ExecutorService executor = Executors.newCachedThreadPool();
	
	public static EventBus getInstance(){
		if(instance==null){
			instance=new EventBus();
		}
		return instance;
	}
	
	private EventBus(){
	}
	
	/**
	 * 添加事件监听器。
	 * @param listener
	 */
	public void addListener(EventListener listener) {
		if (listeners.get(allFilter) == null) {
			synchronized (this) {
				if (listeners.get(allFilter) == null) {
					listeners.put(allFilter, new ArrayList<EventListener>());
				}
			}
		}
		listeners.get(allFilter).add(listener);
	}

	/**
	 * 添加事件监听器，监听器只监听filter所过滤的事件。
	 * @param filter
	 * @param listener
	 */
	public void addListener(EventFilter filter, EventListener listener) {
		if (listeners.get(filter) == null) {
			synchronized (this) {
				if (listeners.get(filter) == null) {
					listeners.put(filter, new ArrayList<EventListener>());
				}
			}
		}
		listeners.get(filter).add(listener);
	}
	
	public void removeEventListener(EventListener listener){
		listeners.get(allFilter).remove(listener);
	}
	
	public void publishEvent(final Event event){
		List<EventListener> interestedListeners = getInterestedListeners(event);
		for (final EventListener listener : interestedListeners) {
			try {
				FutureTask<String> future = new FutureTask<String>(
					new Callable<String>() {
						public String call() {
							listener.handlerEvent(event);
							return null;
						}
					});
				executor.execute(future);
				try {
					//一个事件监听器的处理时间最长为5秒
					future.get(timeout, TimeUnit.MILLISECONDS);
				} catch (InterruptedException e) {
					future.cancel(true);
				} catch (ExecutionException e) {
					future.cancel(true);
				} catch (TimeoutException e) {
					future.cancel(true);
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
	
	/**
	 * 通过对此事件感兴趣的监听器列表。
	 * 
	 * @param event
	 * @return
	 */
	private List<EventListener> getInterestedListeners(Event event) {
		List<EventListener> interestedListeners = new ArrayList<EventListener>();
		for (Entry<EventFilter, List<EventListener>> entry : listeners
				.entrySet()) {
			if (entry.getKey().match(event)) {
				interestedListeners.addAll(entry.getValue());
			}
		}
		return interestedListeners;
	}
	
	/**
	 * 无条件的过滤器，用于监听所有事件。
	 * 
	 * @author Dream.Lee
	 * 
	 */
	private static class AllEventFilter implements EventFilter {

		@Override
		public boolean match(Event event) {
			return true;
		}
	}

}
